package com.rem.flink.flink10Sql;

import com.rem.flink.flink2Source.Event;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 将datastream 转换为 动态表/虚拟表
 * 对表进行输出的动作时分为 toDataStream   toChangelogStream
 * toDataStream：只对插入的数据进行转换，对数据进行增删改操作后再转换会报错
 * toChangelogStream：转换日志流 对数据进行了增删改操作后 需要此方法进转换行操作
 *
 * @author Rem
 * @date 2022-11-07
 */

public class TableToStreamTest {

    public static void main(String[] args) throws Exception {
        // 获取流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据源
        SingleOutputStreamOperator<Event> eventStream = env
                .fromElements(
                        new Event("Alice", "./home", 1000L),
                        new Event("Bob", "./cart", 1000L),
                        new Event("Alice", "./prod?id=1", 5 * 1000L),
                        new Event("Cary", "./home", 60 * 1000L),
                        new Event("Bob", "./prod?id=3", 90 * 1000L),
                        new Event("Alice", "./prod?id=7", 105 * 1000L)
                );

        // 获取表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.1 将数据流转换成 动态表
        Table table = tableEnv.fromDataStream(eventStream);

        //2.2 将数据流转换成虚拟表
        tableEnv.createTemporaryView("eventTable", eventStream);

        //3 查询Alice 的访问 url列表
        Table result1 = tableEnv.sqlQuery(" select user,url from eventTable where user = 'Alice' ");
        //3.1 或者结构化语言查询
        Table result2 = table.select($("user"), $("url")).where($("user").isEqual("Alice"));

        //4 将表转换成数据流，在控制台打印   转换为数据流  只有插入的动作 没有增删改
        tableEnv.toDataStream(result1).print("result1 ");
        tableEnv.toDataStream(result2).print("result2 ");

        //5 统计每个用户的点击次数
        Table result3 = tableEnv.sqlQuery("select user, COUNT(url) FROM eventTable GROUP BY user");

        //6 将表转换成数据流，在控制台打印   转换为日志流  在统计的过程存在增删改
        tableEnv.toChangelogStream(result3).print("result3 ");

        env.execute();
    }
}
